package com.lemon.exam.common.service.impl;

import co.elastic.clients.elasticsearch.indices.CreateIndexResponse;
import co.elastic.clients.transport.endpoints.BooleanResponse;
import com.lemon.exam.common.exception.CustomException;
import com.lemon.exam.common.service.ElasticsearchService;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.elasticsearch.client.elc.ReactiveElasticsearchClient;
import org.springframework.data.elasticsearch.core.ReactiveElasticsearchOperations;
import org.springframework.data.elasticsearch.core.ReactiveIndexOperations;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.util.retry.Retry;

import java.time.Duration;
import java.util.Arrays;

/**
 * ES实现类
 *
 * @author Lemon
 * @since 2025/4/25 17:23
 */
@Slf4j
@Service
public class ElasticsearchServiceImpl implements ElasticsearchService {
    @Resource
    private ReactiveElasticsearchClient client;
    @Resource
    private ReactiveElasticsearchOperations operations;

    @Override
    public Mono<Void> createIndices(String... indexName) {
        return Flux.fromArray(indexName)
                .flatMap(index -> createSingleIndex(index).onErrorResume(e -> Mono.empty()))
                .then()
                .doOnSubscribe(__ -> log.debug("Starting to create indices: {}", Arrays.toString(indexName)))
                .onErrorResume(e -> Mono.error(new CustomException(e.getMessage())));
    }

    /**
     * 创建单个索引
     *
     * @param indexName
     * @return
     */
    private Mono<CreateIndexResponse> createSingleIndex(String indexName) {
        return Mono.defer(() -> client.indices().create(builder -> builder.index(indexName)))
                .doOnSuccess(response -> {
                    if (!response.acknowledged()) {
                        log.warn("索引创建未确认: {}", indexName);
                    }
                    log.info("成功创建索引: {}", indexName);
                })
                .retryWhen(Retry.backoff(3, Duration.ofMillis(100))
                        .doBeforeRetry(retry -> log.warn("正在重试索引创建 (第{}次): {}", retry.totalRetries() + 1, indexName))
                        .onRetryExhaustedThrow((retryBackoffSpec, retrySignal) -> new CustomException("重试后创建索引失败: " + indexName)));
    }

    @Override
    public Mono<Boolean> indexExists(String indexName) {
        return client.indices()
                .exists(builder -> builder.index(indexName))
                .map(BooleanResponse::value);
    }

    @Override
    public Mono<Void> createIndexIfNotExists(String indexName) {
        return indexExists(indexName)
                .flatMap(exists -> exists ? Mono.empty() : createSingleIndex(indexName).then());
    }

    @Override
    public Mono<Boolean> createIndexIfNotExists(Class<?> entity) {
        ReactiveIndexOperations indexOps = operations.indexOps(entity);
        return indexOps.exists()
                .flatMap(exists -> {
                    if (!exists) {
                        return indexOps.createWithMapping();
                    }
                    return indexOps.putMapping();
                });
    }

    @Override
    public Mono<Boolean> deleteAll(String indexName) {
        return client.indices()
                .delete(index -> index.index(indexName))
                .flatMap(deleteResponse -> {
                    if (deleteResponse.acknowledged()) {
                        // 索引删除成功后重新创建
                        return client.indices()
                                .create(index -> index.index(indexName))
                                .map(CreateIndexResponse::acknowledged);
                    }
                    return Mono.just(false);
                });
    }
}
